package cn.codeartist.springboot.component.rocketmq.proxy.impl;

import cn.codeartist.springboot.component.rocketmq.callback.SuccessSendCallback;
import cn.codeartist.springboot.component.rocketmq.proxy.MessageEntity;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Collection;

/**
 * RocketMQ生产者发送消息封装类
 *
 * @author 艾江南
 * @date 2020/8/13
 */
public class DefaultRocketMQProducer extends AbstractRocketMQProducer {

    /*
     * Single
     */

    @Override
    public SendResult send(MessageEntity msg) {
        try {
            return defaultMQProducer.send(toMessage(msg), msg.getTimeout());
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
        return null;
    }

    @Override
    public void sendAsync(MessageEntity msg) {
        sendAsync(msg, sendResult -> {
        });
    }

    @Override
    public void sendAsync(MessageEntity msg, SuccessSendCallback sendCallback) {
        try {
            defaultMQProducer.send(toMessage(msg), toSendCallback(msg, sendCallback), msg.getTimeout());
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public void sendOneWay(MessageEntity msg) {
        try {
            defaultMQProducer.sendOneway(toMessage(msg));
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public SendResult send(MessageEntity msg, MessageQueue mq) {
        try {
            return defaultMQProducer.send(toMessage(msg), mq, msg.getTimeout());
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
        return null;
    }

    @Override
    public void sendAsync(MessageEntity msg, MessageQueue mq) {
        sendAsync(msg, mq, sendResult -> {
        });
    }

    @Override
    public void sendAsync(MessageEntity msg, MessageQueue mq, SuccessSendCallback sendCallback) {
        try {
            defaultMQProducer.send(toMessage(msg), mq, toSendCallback(msg, sendCallback), msg.getTimeout());
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public void sendOneWay(MessageEntity msg, MessageQueue mq) {
        try {
            defaultMQProducer.sendOneway(toMessage(msg), mq);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public SendResult send(MessageEntity msg, MessageQueueSelector selector, Object arg) {
        try {
            return defaultMQProducer.send(toMessage(msg), selector, arg, msg.getTimeout());
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
        return null;
    }

    @Override
    public void sendAsync(MessageEntity msg, MessageQueueSelector selector, Object arg) {
        sendAsync(msg, selector, arg, sendResult -> {
        });
    }

    @Override
    public void sendAsync(MessageEntity msg, MessageQueueSelector selector, Object arg, SuccessSendCallback sendCallback) {
        try {
            defaultMQProducer.send(toMessage(msg), selector, arg, toSendCallback(msg, sendCallback));
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public void sendOneWay(MessageEntity msg, MessageQueueSelector selector, Object arg) {
        try {
            defaultMQProducer.sendOneway(toMessage(msg), selector, arg);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
    }

    @Override
    public TransactionSendResult sendMessageInTransaction(MessageEntity msg, LocalTransactionExecuter tranExecutor, Object arg) {
        try {
            return defaultMQProducer.sendMessageInTransaction(toMessage(msg), tranExecutor, arg);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msg, e);
        }
        return null;
    }

    /*
     * Batch
     */

    @Override
    public void send(Collection<MessageEntity> msgs) {
        try {
            defaultMQProducer.send(toMessage(msgs));
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msgs, e);
        }
    }

    @Override
    public void send(Collection<MessageEntity> msgs, long timeout) {
        try {
            defaultMQProducer.send(toMessage(msgs), timeout);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msgs, e);
        }
    }

    @Override
    public void send(Collection<MessageEntity> msgs, MessageQueue mq) {
        try {
            defaultMQProducer.send(toMessage(msgs), mq);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msgs, e);
        }
    }

    @Override
    public void send(Collection<MessageEntity> msgs, MessageQueue mq, long timeout) {
        try {
            defaultMQProducer.send(toMessage(msgs), mq, timeout);
        } catch (Exception e) {
            logger.error("RocketMQ producer {} send message async callback error. Message: {},", beanName, msgs, e);
        }
    }
}
